package com.example.consumerdemo.common;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.MessageListener;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;

/**
 * @author xzc
 */
@Component
public class RocketMQConsumerConfiguration {

    Logger LOGGER = LoggerFactory.getLogger(RocketMQConsumerConfiguration.class);

    @Value("${rocketmq.instanceName}")
    private String instanceName;
    @Value("${rocketmq.namesrvAddr}")
    private String namesrvAddr;
    @Value("${rocketmq.groupName}")
    private String groupName;
    @Value("${rocketmq.topic}")
    private String topic;
    @Value("${rocketmq.tag}")
    private String tag;
    @Value("${rocketmq.consumerSize}")
    private int consumerSize;
    @Value("${rocketmq.consumeThreadMin}")
    private int consumeThreadMin;
    @Value("${rocketmq.consumeThreadMax}")
    private int consumeThreadMax;
    @Value("${rocketmq.consumeMessageBatchMaxSize}")
    private int consumeMessageBatchMaxSize;


    @Bean(value = "testMq")
    public List<DefaultMQPushConsumer> getRocketMQConsumer() throws MQClientException {

        List<DefaultMQPushConsumer> list = new ArrayList<>();
        for (int i = 0; i < consumerSize; i++) {
            String consumerName = instanceName + "-" + i;
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(groupName);
            consumer.setInstanceName(consumerName);
            // 可指定 tag
//            consumer.subscribe(topic, "jd");
            MessageListener messageListener = new com.example.consumerdemo.common.MessageListener(consumerName);
            consumer.registerMessageListener(messageListener);
            // 如果非第一次启动，那么按照上次消费的位置继续消费
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

            startMq(consumer);
            LOGGER.info("消费者" + consumerName + "启动");
            list.add(consumer);
        }
        return list;
    }

    private void startMq(DefaultMQPushConsumer consumer){
        try {
            consumer.setNamesrvAddr(namesrvAddr);
            consumer.setConsumeThreadMin(consumeThreadMin);
            consumer.setConsumeThreadMax(consumeThreadMax);
            consumer.setConsumeMessageBatchMaxSize(consumeMessageBatchMaxSize);
            consumer.subscribe(topic, this.tag);
            // 集群消费
            consumer.setMessageModel(MessageModel.CLUSTERING);
            consumer.start();
            LOGGER.info("consumer is start !!! groupName:{},topic:{},namesrvAddr:{}", groupName, topic, namesrvAddr);
        } catch (MQClientException e) {
            LOGGER.error("consumer is start !!! groupName:{},topic:{},namesrvAddr:{}", groupName, topic, namesrvAddr, e);
        }
    }

}
